package com.javasearch.utils;


import com.google.common.util.concurrent.ThreadFactoryBuilder;

import java.util.Collection;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;
import java.util.function.Consumer;

public class ExecutorUtils {

    private   static final Map<String, ExecutorService> map = new ConcurrentHashMap<>();

    //默认获取线程池    默认任务数量=核心*2  , 任务时间=0.5   ,任务时间是按照秒计算的
    public static ExecutorService getExecutor(String key) {
        return getExecutor(key,Runtime.getRuntime().availableProcessors() ,0.5);
    }
    //核心线程数*任务执行时间=核心线程数 ,最大线程数设置的=核心线程数
    public static ExecutorService getExecutor(String key,double taskTime) {
        return getExecutor(key,Runtime.getRuntime().availableProcessors() ,taskTime);
    }
    //根据任务数据和任务时间获取线程池
    public static ExecutorService getExecutor(String key,int taskSize,double taskTime) {
        ExecutorService executorService = map.get(key);
        if (executorService == null) {
            synchronized (ExecutorUtils.class) {
                executorService = map.get(key);
                if (executorService == null) {
                    executorService = compute(key,taskSize,taskTime);
                }
            }
        }
        return executorService;
    }
    public static ExecutorService getExecutorCustom(String key,int taskSize,int queueSize ) {
        //使用1分钟就行了
        TimeUnit keepAliveTiime = TimeUnit.MINUTES;
        //队列长度 maxPoolSize*2
        BlockingQueue<Runnable> blockingQueue = new ArrayBlockingQueue<>(Math.max(queueSize, 100));
        ThreadFactory namedThreadFactory = new ThreadFactoryBuilder()
                .setNameFormat(key+"-%d").build();
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(
                Runtime.getRuntime().availableProcessors(),
                Math.max(taskSize, Runtime.getRuntime().availableProcessors()),
                1,
                keepAliveTiime,
                blockingQueue,
                namedThreadFactory,
                new ThreadPoolExecutor.CallerRunsPolicy()
        );
        return threadPoolExecutor;
    }


    private static ExecutorService compute(String key,int taskSize,double taskTime){
        int corePoolSize=taskSize; //核心线程数
        int floor = (int) Math.floor((double) taskSize * taskTime);
        int maxPoolSize= Math.max(taskSize, floor); //最大线程数
        //使用1分钟就行了
        TimeUnit keepAliveTiime = TimeUnit.MINUTES;
        //队列长度 maxPoolSize*2
        BlockingQueue<Runnable> blockingQueue = new ArrayBlockingQueue<>((int) (1000*maxPoolSize));
        ThreadFactory namedThreadFactory = new ThreadFactoryBuilder()
                .setNameFormat(key+"-%d").build();
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(
                corePoolSize,
                maxPoolSize,
                1,
                keepAliveTiime,
                blockingQueue,
                namedThreadFactory,
                new ThreadPoolExecutor.CallerRunsPolicy()
        );
        return threadPoolExecutor;
    }





    //创建普通线程

    public static void newThread(Runnable runnable) {
        Thread thread = new Thread(runnable);
        thread.setUncaughtExceptionHandler((Thread t, Throwable e) -> {System.out.println(t.getName() + ": " + e.getMessage());});
        thread.start();
    }
    //守护线程
    public static void newThreadDaemon(Runnable runnable) {
        Thread thread = new Thread(runnable);
        thread.setUncaughtExceptionHandler((Thread t, Throwable e) -> {System.out.println(t.getName() + ": " + e.getMessage());});
        thread.setDaemon(true);
        thread.start();
    }

    public static <T> Future<T> newThread(Callable<T> callable) {
        FutureTask<T> futureTask = new FutureTask<T>(callable);
        Thread thread = new Thread(futureTask);
        thread.setUncaughtExceptionHandler((Thread t, Throwable e) -> {System.out.println(t.getName() + ": " + e.getMessage());});
        thread.start();
        return futureTask;
    }

    public static void create(String key, Runnable runnable) {
        getExecutor(key).submit(runnable);
    }
    //创建线程池,运行指定任务,并且指定线程数量
    public static void create(String key,int taskSize,double taskTime, Runnable runnable) {
        getExecutor(key,taskSize,taskTime).submit(runnable);
    }
    public static void create(String key,double taskTime, Runnable runnable) {
        getExecutor(key,taskTime).submit(runnable);
    }

    public static void createAndMaximumPoolSize(String key, List<Runnable> executorFunction) {
        for (Runnable function : executorFunction) {
            getExecutor(key).submit(function);
        }
    }

    public static Future<?> createFutureR(String key, Callable callable) {
        return getExecutor(key).submit(callable);

    }

    public static Future<?> createFuture(String key, Runnable runnable) {
        return getExecutor(key).submit(runnable);

    }


    public static Collection<Future<?>> createFutures(String key, List<Runnable> runnables) {
        Collection<Future<?>> futures = new LinkedList<Future<?>>();
        for (Runnable function : runnables) {
            futures.add(getExecutor(key).submit(function));
        }
        return futures;
    }



    public static  Collection<Future<?>> createFuturesR(String key, List<Callable> callables) {
        Collection<Future<?>> futures = new LinkedList<Future<?>>();
        for (Callable callable : callables) {
            futures.add(getExecutor(key).submit(callable));
        }
        return futures;
    }
    //不用等待所有线程执行完毕,而是谁先执行完毕,就返回谁的结果 ,以此类推,等待全部线程执行完毕
    public  static void createCompletionServicesAll(String key, List<Callable> callables, Consumer< Future> consumer)  {
        CompletionService<?> completionService = new ExecutorCompletionService<>(getExecutor(key));
        for (Callable callable : callables) {
            completionService.submit(callable);
        }
        try {
            for (int i = 0; i < callables.size(); i++) {
                consumer.accept(completionService.take());
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
    //一堆线程同时执行,谁先执行完毕那么就采用谁的结果,其余线程结果不管  ,注意如果执行了,
    // 但是没有任何线程结果返回,那么一直等待,如果是线程内部向外部报错了,那么也算返回了,并且打印错误
    public  static Object createCompletionServicesOne(String key, List<Callable> callables) throws ExecutionException {
        CompletionService<?> completionService = new ExecutorCompletionService<>(getExecutor(key));
        for (Callable callable : callables) {
            completionService.submit(callable);
        }
        Object result=null;
        try {
            result=completionService.take().get();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        return result;
    }




    //阻塞全部线程完成后
    public static void waitComplete(Collection<Future<?>> collection) {
        for (Future<?> future : collection) {
            try {
                future.get();
            } catch (InterruptedException | ExecutionException e) {
                e.printStackTrace();
            }
        }
    }
    public static void waitComplete(Future<?> future) {
        try {
            future.get();
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }
    }

    //关闭线程池
    public static void shutdown(String key) {
        ExecutorService executorService = map.get(key);
        if (executorService != null) {
            executorService.shutdown();
            map.remove(key);
        }
    }


}
